package com.atguigu.flink.timeAndwindow;

import com.atguigu.flink.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * Created by 黄凯 on 2023/6/16 0016 20:14
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 */
public class Flink03_WaterMarkTransfer {

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 5678);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        env.disableOperatorChaining();

        SingleOutputStreamOperator<Event> ds1 = env.socketTextStream("127.0.0.1", 8888)
                .map(

                        line -> {

                            String[] fields = line.split(",");
                            return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));

                        }

                ).assignTimestampsAndWatermarks(

                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (element, recordTimestamp) -> element.getTs()
                                )

                );

        SingleOutputStreamOperator<Event> ds2 = env.socketTextStream("127.0.0.1", 9999)
                .map(

                        line -> {

                            String[] fields = line.split(",");
                            return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));

                        }

                ).assignTimestampsAndWatermarks(

                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (element, recordTimestamp) -> element.getTs()
                                )

                );

        /**
         * 上游给下游选择广播水位线
         */
       /* ds1.map(x -> x)
                .setParallelism(2)
                .print();*/


        /**
         * 多个上游给同一个下游传递水位线，下游取最小
         */
        ds1.map(x -> x)
                .union(ds2.map(x -> x))
                .map(x -> x)
                .print();

        env.execute();

    }

}
